package db.nio.framer;

import db.nio.Framer;
import sun.net.ConnectionResetException;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeoutException;

/**
 * 带长度报头协议封装类
 * Created by daibin on 10/20/16.
 */
public class LengthFieldFramer implements Framer {

    /**
     * 通道事件触发器
     */
    private SelectionKey selectionKey;


    public LengthFieldFramer() {
    }


    @Override
    public void setSelectionKey(SelectionKey key) {
        this.selectionKey = key;
    }


    /**
     * 读出缓冲区中所有数据
     * @return
     * @throws Exception
     */
    @Override
    public List<ByteBuffer> read() throws Exception{

        List<ByteBuffer> byteBufferList = new ArrayList<ByteBuffer>();
        SocketChannel channel = (SocketChannel) selectionKey.channel();

        Selector selector = null;
        try {
            //开启监听，接收剩下包
            selector = Selector.open();
            channel.register(selector, SelectionKey.OP_READ);
            int nKey = selector.select(1);
            while (nKey > 0) {
                Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                while (keys.hasNext()) {
                    SelectionKey key = keys.next();
                    keys.remove();
                    if(key.isReadable()){ //进行读数据
                        byteBufferList.add(read(selector,key));
                    }
                }

                nKey = selector.select(1);
            }
        } finally {
            if (selector != null) {
                SelectionKey key = channel.keyFor(selector);
                if (key != null)
                    key.cancel();
                if (selector.isOpen())
                    selector.close();
            }
        }
        return byteBufferList;
    }


    /**
     * 读单个数据
     * @param selector
     * @param selectionKey
     * @return
     * @throws Exception
     */
    private ByteBuffer read(Selector selector,SelectionKey selectionKey) throws Exception{
        SocketChannel channel = (SocketChannel) selectionKey.channel();
        //读取报头
        ByteBuffer headBuffer = ByteBuffer.allocate(4);
        int maxLen; //包总长度
        try {
            try {
                while (channel.read(headBuffer) > 0) ;
            } finally {
                headBuffer.flip();
            }

            //接收数据为空，表示连接关闭或重置
            if(headBuffer.limit() == 0)
                throw new ConnectionResetException("connect reset");

            if (headBuffer.limit() < 4)
                throw new IOException("read header error");

            maxLen = headBuffer.getInt();

        } finally {
            headBuffer.clear();
        }

        ByteBuffer buffer = ByteBuffer.allocate(maxLen);
        try {
            //读取数据
            while (channel.read(buffer) > 0) ;

            //未读满则等待接收完整数据
            if (buffer.position() < buffer.limit()) {
                this.readComplete(selector,selectionKey,buffer);
            }
        }finally {
            buffer.flip();
        }
        return buffer;
    }



    /**
     * 读取完整
     * @param buffer
     * @throws Exception
     */
    private void readComplete(Selector selector,SelectionKey selectionKey,ByteBuffer buffer)throws Exception {
        int timeOut = 0;
        int count = 0;
        while (buffer.position() < buffer.limit()) {
            long start = System.currentTimeMillis();
            int nkey = selector.select(1);

            //超时处理
            if (timeOut++ > 3000)
                throw new TimeoutException("read time out");

            if (nkey > 0) {
                timeOut = 0;
                Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                while (keys.hasNext()) {
                    timeOut = 0;
                    count = 0;
                    SelectionKey key = keys.next();
                    keys.remove();
                    if (key.isReadable()) {
                        SocketChannel channel = (SocketChannel) key.channel();
                        while (channel.read(buffer) > 0) ;
                    }
                }
            }

            long split = System.currentTimeMillis() - start;
            if (split < 1) {
                if (count++ > 100) {
                    count = 0;

                    SocketChannel channel = (SocketChannel) selectionKey.channel();
                    //替换selector
                    Object attch = selectionKey.attachment();
                    selectionKey.cancel();
                    selector.selectNow();
                    selector.close();

                    Selector newSelector = Selector.open();
                    channel.register(newSelector, SelectionKey.OP_READ, attch);
                    selector = newSelector;
                    selector.wakeup();
                    System.out.println("selector reset");
                }
            }
        }
    }

    @Override
    public void write(ByteBuffer buffer) throws Exception {

        //通道
        SocketChannel channel = (SocketChannel) selectionKey.channel();

        //写空间大小
        int bufferSize = channel.socket().getSendBufferSize();

        //报头
        ByteBuffer bufferHeader = ByteBuffer.allocate(8);
        bufferHeader.putInt(buffer.limit());
        bufferHeader.flip();
        channel.write(bufferHeader);

        //发送消息
        if (buffer.limit() > bufferSize) {

            int currentLen = bufferSize;
            byte[] data = buffer.array();
            ByteBuffer tempBuffer = ByteBuffer.allocate(bufferSize);
            int offset = 0;

            Selector selector = Selector.open();
            channel.register(selector, SelectionKey.OP_WRITE);

            int timeOut = 0;
            while (offset < buffer.limit()) {

                //超时处理
                if (timeOut++ > 3000) {
                    throw new TimeoutException("write time out");
                }

                int nkey = selector.select(1);
                if (nkey > 0) {
                    timeOut = 0;
                    Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                    while (keys.hasNext()) {
                        SelectionKey key = keys.next();
                        keys.remove();
                        if (key.isWritable()) {
                            timeOut = 0;
                            try {
                                tempBuffer.put(data, offset, currentLen);
                                tempBuffer.flip();
                                channel.write(tempBuffer);
                            } finally {
                                tempBuffer.clear();
                            }

                            offset = offset + currentLen;
                            //记录下一个包长度
                            if (offset + bufferSize > data.length) {
                                currentLen = data.length - offset;
                            } else {
                                currentLen = bufferSize;
                            }

                        }
                    }
                }
            }
        } else {
            channel.write(buffer);
        }
    }

    @Override
    public void close() throws IOException {
        try {
            selectionKey.channel().close();
            selectionKey.cancel();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
